package com.foreveross.fetl.queue;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.annotation.Resource;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.log4j.Logger;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

import com.foreveross.fetl.common.HBaseTable;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.vo.FetlErrorMessage;
import com.foreveross.fetl.core.domain.single.SinglePlaneInfoEntity;
import com.foreveross.fetl.core.domain.task.SourceEntity;
import com.foreveross.fetl.core.domain.task.TaskTemplateEntity;
import com.foreveross.fetl.queue.util.ExchangeRateUtil;
import com.foreveross.fetl.queue.util.LogMessageUtils;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.FetlErrorUtils;
import com.foreveross.fetl.util.HbaseToPlaneInfoUtil;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.util.StringUtils;

/*-
 *  
 *  触发时间： 每小时触发一次
 * 
 * 
 *抓取基础数据表(CRAWL_PLANE_FREIGHT)：  
 *rowkey设计
 *抓取基础数据表(CRAWL_PLANE_FREIGHT)：  
 *抓取日期(年月日时)_渠道_区域ID_起点(使用三字码)_终点(使用三字码)_航班日期(yyyymmdd)_航班ID_5位随即数
 8位     +   5位   +2位   +       3位      +  3 位  +   8位   +          7位       + 5位
 *     
 *    for(每小时抓取的任务){
 *     	  for(每条航班){
 *     		 1.每小时的[官网、渠道]航班最低价LP_FLIGHT    rowkey： 渠道_起点(使用三字码)_终点(使用三字码)_(航班号)_航班日期(年月日时)
 *           2.每小时的[官网、渠道]航线最低价LP_LINE_HOUR      rowkey： 渠道_起点(使用三字码)_终点(使用三字码)_航班日期(年月日时)
 *     		 3.每小时的[官网、渠道]航班仓位最低价LP_FLIGHT_POS_HOUR rowkey： 渠道_起点(使用三字码)_终点(使用三字码)_(航班号)_仓位_航班日期(年月日时)  
 *        }
 *     }
 *     
 *     for(每天的按小时统计结果){
 *     		 1.每天的[官网、渠道]航班最低价LP_FLIGHT_DAY    rowkey： 渠道_起点(使用三字码)_终点(使用三字码)_(航班号)_航班日期(年月日)
 *           2.每天的[官网、渠道]航线最低价LP_LINE_DAY      rowkey： 渠道_起点(使用三字码)_终点(使用三字码)_航班日期(年月日)
 *     		 3.每天的[官网、渠道]航班仓位最低价LP_FLIGHT_POS_DAY rowkey： 渠道_起点(使用三字码)_终点(使用三字码)_(航班号)_仓位_航班日期(年月日)  
 *     }
 *     
 *     官网渠道价差分析：
 *     航班周差价3%、6%统计
 *     
 *     
 */

//@Transactional
/**
 * 统计频率：每小时统计一次
 * 
 * @author tss
 * 
 */
@Component
public class LastestDataPreprocess {

	private static Logger log = Logger.getLogger("last");
	
	@Resource(name="taskExecutor")
	private  TaskExecutor taskExecutor;
	
	private int tasksNumber;//任务数 多少条rowkey
	
	private int failuresNumber;//失败数
	
	private  int  threadsNumber;//线程数目
	
	private int executeNumber;//已执行的任务条数
	
	
	private int insertNumber; //插入条数 
	
	private long excuteStartTime;
	
	private String taskTime=null;
	
	

	public void execute() {
		Date endTime=new Date();
		rollBackNumber();
		Calendar c = Calendar.getInstance();
		c.add(Calendar.HOUR_OF_DAY, -1);
		taskTime=DateUtils.format(c.getTime(), "yyyy-MM-dd HH");
		process(c.getTime(), endTime);
	}
	
	public void  rollBackNumber(){
		tasksNumber=0;
		failuresNumber=0;
		threadsNumber=0;
		executeNumber=0;
		insertNumber=0;
		excuteStartTime = System.currentTimeMillis();
		LogMessageUtils.create("实时价格处理", this.getClass().getSimpleName());
	}
	
	public void manualExecute(final String start,String time){
		this.rollBackNumber();
		try {
		List<String> timeList = Arrays.asList(time.split(";"));
		Calendar c = Calendar.getInstance();
		taskTime = start + " " +time;
		for (String s : timeList) {
			String str =start;
			str += s;
			Date startTime = DateUtils.parse(str, "yyyyMMddHH");
			c.setTime(startTime);
			c.add(Calendar.HOUR_OF_DAY, +1);
			Date endTime = c.getTime();
			this.process(startTime, endTime);
		}
		} catch (Exception e) {
			LogMessageUtils.build("实时价格处理", this.getClass().getSimpleName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
					insertNumber, taskTime, excuteStartTime);
			log.error(StringUtils.getStackTrace(e));
		}
	}

	/**
	 * 启动处理任务
	 * 
	 * @param start
	 *            格式： yyyyMMddHH
	 * @param end
	 *            格式： yyyyMMddHH
	 */
	public void process(final Date start, final Date end) {
	//	long startL = System.currentTimeMillis();
		log.info("LastestDataPreprocess start:"
				+ DateUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"));
		SourceEntity source = new SourceEntity();
		List<SourceEntity> listSource=source.findAll();
		List<String> strchannels= new ArrayList<String>();
		if(null!=listSource){
			for(SourceEntity entity:listSource){
				strchannels.add(entity.getHbaseKey());
			}
		}
		//重新初始化汇率
		ExchangeRateUtil.initExchangeRate(DateUtils.format(new Date(), "yyyyMMdd"));
		
//		for (final String channel : strchannels) {
//				threadsNumber++;
//		}
		threadsNumber=strchannels.size();
		for (final String channel : strchannels) {
				
				taskExecutor.execute(new Runnable() {
					@Override
					public void run() {
					//	log.info("start channel:"+channel+" area:"+area);
						Calendar cStart = Calendar.getInstance();
						cStart.setTime(start);
						Calendar cEnd = Calendar.getInstance();
						cEnd.setTime(end);
						while (cStart.before(cEnd)) {
							processDataByHour(cStart.getTime(), channel);//20130319142000401
							cStart.add(Calendar.HOUR_OF_DAY, 1);// 开始时间加一小时
						}
					//	log.info("end channel："+channel+" area："+area);
					}
				});
			
		}
//		log.info("LastestDataPreprocess end ："
//				+ DateUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS")+" spend time:"+(System.currentTimeMillis() - startL));

	}

	/*-
	 * 
	 * 充分利用hbase数据库rowkey的字典序及rowkey的设计，将相同的航班放在相邻的位置进行统计，减少内存的占用
	 * 
	 * 按小时处理每小时的最低价
	 * 	 BASE_DATA表的 rowkey设计
	 *	 抓取基础数据表(BASE_DATA)：
	  *	 抓取日期(年月日时)_渠道_区域ID_起点(使用三字码)_终点(使用三字码)_航班日期(年月日时)_航班ID_5位随即数
 *	         10位 +    5位  + 2位 +    3位   +         3位      +           10位+            8位 +    5位
	 * @param startRowKey
	 * @param endRowKey
	 * @throws IOException
	 * @throws Exception
	 */
	private void processDataByHour( Date time ,String channel) {
		ResultScanner rs = null;
		TaskTemplateEntity task=new TaskTemplateEntity();
		List<TaskTemplateEntity> listTask= task.findGralChannelAirLine(channel);
		String startRow=RowkeyUtils.getRowKey(time, channel,  RowkeyUtils.START_ROW_KEY);
		String endRow=RowkeyUtils.getRowKey(time, channel,  RowkeyUtils.STOP_ROW_KEY);
		try {
			
			if(null==listTask || listTask.isEmpty()){
				rs = HBaseHelper.getRecords(HBaseTable.BASE_DATA_FLIGHT,
						startRow, endRow);
				this.processHbaseData(rs,channel,null);
			}else{
				Set<String> fromCitySet=new HashSet<String>();
				for(TaskTemplateEntity taskTem:listTask){
					fromCitySet.add(taskTem.getFromCity());
				}
				for(String fromCity:fromCitySet){
					try{
						startRow=RowkeyUtils.getRowKey(DateUtils.format(time, "yyyyMMddHH"), channel,fromCity, null, RowkeyUtils.START_ROW_KEY);
						endRow=RowkeyUtils.getRowKey(DateUtils.format(time, "yyyyMMddHH"), channel,fromCity, null, RowkeyUtils.STOP_ROW_KEY);
						rs = HBaseHelper.getRecords(HBaseTable.BASE_DATA_FLIGHT,
								startRow, endRow);
//						log.error(startRow);
//						log.error(endRow);
						this.processHbaseData(rs,channel,fromCity);
					}catch (Exception e1) {
						try{
							Thread.sleep(10 * 1000L);
							rs = HBaseHelper.getRecords(HBaseTable.BASE_DATA_FLIGHT,
									startRow, endRow);
							this.processHbaseData(rs,channel,fromCity);
						}catch (Exception e) {
							failuresNumber++;
							log.error(startRow);
							log.error(endRow);
							log.error(StringUtils.getStackTrace(e));
							FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.last, this.getClass().getSimpleName(),DateUtils.format(time, "yyyyMMddHH"), e1);
							vo.setStartRowKey(startRow);
							vo.setEndRowkey(endRow);
							FetlErrorUtils.saveToHbase(vo);
						}
					}
				}
			}
		} catch (Exception e1) {
			log.error(StringUtils.getStackTrace(e1));
			String str=DateUtils.format(time, "yyyyMMddHH");
			FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.last, this.getClass().getSimpleName(),str, e1);
			FetlErrorUtils.saveToHbase(vo);
		}finally{
			if(rs != null) rs.close();
			rs = null;
			executeNumber++;
			log.error(DateUtils.format(time, "ddHH")+"   "+channel+ "   "+executeNumber);
			if(executeNumber==threadsNumber){
				LogMessageUtils.build("实时价格处理", this.getClass().getSimpleName(), 2, null, 
						failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime);
			}
		}
		Thread.interrupted();
	}
	
	private void processHbaseData(ResultScanner rs, String channel, String fromCity) {
		Set<SinglePlaneInfoEntity> setPlaneInfo=new HashSet<SinglePlaneInfoEntity>(1000);
		SinglePlaneInfoEntity singlePlaneInfo =null;
		String id =null;
		Date grabTime=null;
		for (Result res : rs) {
			try {// 某一条航班数据的错误，不应该影响其他航班数据的预处理
//			
				id=new String(res.getRow());
				tasksNumber++;
				Map<String, Map<String, String>> curMap = HBaseHelper
						.transformHBaseRowToMap(res);
				
				Map<String, String> curPalneInfoMap = curMap.get("planeinfo");
				grabTime=DateUtils.parse(curPalneInfoMap.get("grabDate"), "yyyyMMddHHmmssSSS");
				/*
				float flightPrice = Float.parseFloat(curPalneInfoMap.get("lowerPrice") != null 
						 ? curPalneInfoMap.get("lowerPrice") : "0");
				if (flightPrice <= 0) {
					if(flightPrice <0){
						log.error(id + "  lowerPrice is " + flightPrice);
					}
					continue;
				}*/
				singlePlaneInfo=this.hbaseToEntity(curMap, id);
				if(null!=singlePlaneInfo){
					setPlaneInfo.add(singlePlaneInfo);
				}
			} catch (Exception e) {
				failuresNumber++;
				log.error(id);
				log.error(StringUtils.getStackTrace(e));
				FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.last, this.getClass().getSimpleName(),DateUtils.format(grabTime, "yyyyMMddHH"), e);
				List<String> list =new ArrayList<String>();
				list.add(id);
				vo.setKeys(list);
				FetlErrorUtils.saveToHbase(vo);
			}
		
		}
		List<SinglePlaneInfoEntity> oldList =this.findOldEngity(setPlaneInfo,channel,fromCity);
		this.saveToMysql(setPlaneInfo,oldList);
	}
	
	/**
	 * 先删除旧的 oldList 再保存新的数据setPlaneInfo
	 * @param setPlaneInfo
	 * @param oldList
	 */
	private void saveToMysql(Set<SinglePlaneInfoEntity> setPlaneInfo,
			List<SinglePlaneInfoEntity> oldList) {
		SinglePlaneInfoEntity planInfo=new SinglePlaneInfoEntity();
		try {
			
			if(null!=oldList && oldList.size()>0){
				planInfo.delete(oldList);
			}
		} catch (SQLException e) {
			log.error("delete  planeInfo error");
			log.error(StringUtils.getStackTrace(e));
		}
			try {
				if(null!=setPlaneInfo && setPlaneInfo.size()>0){
					int size=0;
					Set<SinglePlaneInfoEntity> set = new HashSet<SinglePlaneInfoEntity>(1024);
					for(SinglePlaneInfoEntity entity:setPlaneInfo){
						set.add(entity);
						size++;
						if(size==1000){
							planInfo.create(set);
							size=0;
							set.clear();
						}
					}
					if(size>0){
						planInfo.create(set);
						set.clear();
					}
					insertNumber+=setPlaneInfo.size();
				}
				 
			} catch (SQLException e) {
				log.error("create  planeInfo error");
				log.error(StringUtils.getStackTrace(e));
				failuresNumber+=setPlaneInfo.size();
			}
		
	}

	/**
	 * 找出旧数据里与新数据相同的集合
	 * @param setPlaneInfo
	 * @param channel 
	 * @param fromCity 
	 * @return
	 */
	private List<SinglePlaneInfoEntity> findOldEngity(Set<SinglePlaneInfoEntity> setPlaneInfo, String channel, String fromCity) {
		SinglePlaneInfoEntity singlePlaneInfo =new SinglePlaneInfoEntity();
		if(null==setPlaneInfo || setPlaneInfo.isEmpty()) 
			return null;
		
		List<SinglePlaneInfoEntity> list=null;
		try {
			list=singlePlaneInfo.findByCondition(channel, fromCity, null, null, null);
		} catch (SQLException e) {
			log.error(StringUtils.getStackTrace(e));
		}
		if(null!= list && list.size()>0 ){
			List<SinglePlaneInfoEntity> oldList =new ArrayList<SinglePlaneInfoEntity>();
			for(SinglePlaneInfoEntity newEntity:setPlaneInfo){
				for(SinglePlaneInfoEntity oldEntity:list){
					if(newEntity.equals(oldEntity)){
						oldList.add(oldEntity);
						list.remove(oldEntity);
						break;
					}
				}
			}
			if(null!=oldList && oldList.size()>0){
				return oldList;
			}
		}
		return null;
	}

	/**
	 * 将hbase 的一条记录转换为 SinglePlaneInfoEntity 对象
	 * @param fcv
	 * @param id
	 * @return
	 */
	private SinglePlaneInfoEntity hbaseToEntity(Map<String, Map<String, String>> fcv,String id){
		SinglePlaneInfoEntity singlePlaneInfo =null;
		double refePrice=ExchangeRateUtil.getRefePrice(fcv.get("planeinfo").get("currency"));

		singlePlaneInfo = HbaseToPlaneInfoUtil.hbaseToSinglePlaneInfo(fcv,refePrice,SinglePlaneInfoEntity.class);
		if(null==singlePlaneInfo || null==singlePlaneInfo.getAttachHbaseKey() ||
				null==singlePlaneInfo.getFromCity() || 	 null==singlePlaneInfo.getToCity() ||
				null==singlePlaneInfo.getFlightDate() || null==singlePlaneInfo.getFlightNo()) {
			log.error(id);
			if (null == singlePlaneInfo) {
				log.error("singlePlaneInfo is null");
			} else if (null == singlePlaneInfo.getAttachHbaseKey()) {
				log.error("attachHbaseKey is null");
			} else if (null == singlePlaneInfo.getFromCity()) {
				log.error("fromCity is null");
			} else if (null == singlePlaneInfo.getToCity()) {
				log.error("toCity is null");
			} else if (null == singlePlaneInfo.getFlightDate()) {
				log.error("flightDate is null");
			} else if (null == singlePlaneInfo.getFlightNo()) {
				log.error("flightId is null");
			}
			return null;
		}
		
		//价格为0的不要取
		if(null==singlePlaneInfo.getTotalLowestPrice()|| singlePlaneInfo.getTotalLowestPrice()==0){
			return null;
		}
		
		singlePlaneInfo.setRowkey(id);
		return singlePlaneInfo;
	}

	
}
